package com.ruoyi.teaching.service.impl;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

import com.ruoyi.common.core.exception.ServiceException;
import com.ruoyi.common.core.utils.DateUtils;
import com.ruoyi.teaching.domain.YaCourseDesignInterview;
import com.ruoyi.teaching.flink.BatchESSink;
import com.ruoyi.teaching.flink.MySQLSource;
import com.ruoyi.teaching.mapper.YaCourseDesignInterviewMapper;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.elasticsearch.client.elc.ElasticsearchTemplate;
import org.springframework.stereotype.Service;
import com.ruoyi.teaching.mapper.YaHighInterviewMapper;
import com.ruoyi.teaching.domain.YaHighInterview;
import com.ruoyi.teaching.service.IYaHighInterviewService;

/**
 * 高频面试题Service业务层处理
 *
 * @author ruoyi
 * @date 2025-09-03
 */
@Service
public class YaHighInterviewServiceImpl implements IYaHighInterviewService {
  @Autowired
  private YaHighInterviewMapper yaHighInterviewMapper;

  @Autowired
  private YaCourseDesignInterviewMapper yaCourseDesignInterviewMapper;

  @Autowired
  private ElasticsearchTemplate elasticsearchTemplate;

  @Autowired
  private RabbitTemplate rabbitTemplate;

  /**
   * 查询高频面试题
   *
   * @param id 高频面试题主键
   * @return 高频面试题
   */
  @Override
  public YaHighInterview selectYaHighInterviewById(Long id) {
    return yaHighInterviewMapper.selectYaHighInterviewById(id);
  }

  /**
   * 查询高频面试题列表
   *
   * @param yaHighInterview 高频面试题
   * @return 高频面试题
   */
  @Override
  public List<YaHighInterview> selectYaHighInterviewList(YaHighInterview yaHighInterview) {
    return yaHighInterviewMapper.selectYaHighInterviewList(yaHighInterview);
  }

  /**
   * 新增高频面试题
   *
   * @param yaHighInterview 高频面试题
   * @return 结果
   */
  @Override
  public int insertYaHighInterview(YaHighInterview yaHighInterview) {
    yaHighInterview.setCreateTime(DateUtils.getNowDate());
    int i = yaHighInterviewMapper.insertYaHighInterview(yaHighInterview);
//    rabbitTemplate.convertAndSend("teaching.interview.ex","teaching.interview.ex.add",yaHighInterview);
    return i;
  }

  /**
   * 修改高频面试题
   *
   * @param yaHighInterview 高频面试题
   * @return 结果
   */
  @Override
  public int updateYaHighInterview(YaHighInterview yaHighInterview) {
    yaHighInterview.setUpdateTime(DateUtils.getNowDate());
    int i = yaHighInterviewMapper.updateYaHighInterview(yaHighInterview);
//    rabbitTemplate.convertAndSend("teaching.interview.ex","teaching.interview.ex.update",yaHighInterview);
    return i;
  }

  /**
   * 批量删除高频面试题
   *
   * @param ids 需要删除的高频面试题主键
   * @return 结果
   */
  @Override
  public int deleteYaHighInterviewByIds(Long[] ids) {
    return yaHighInterviewMapper.deleteYaHighInterviewByIds(ids);
  }

  /**
   * 删除高频面试题信息
   *
   * @param id 高频面试题主键
   * @return 结果
   */
  @Override
  public int deleteYaHighInterviewById(Long id) {
    int i = yaHighInterviewMapper.deleteYaHighInterviewById(id);
//    rabbitTemplate.convertAndSend("teaching.interview.ex","teaching.interview.ex.delete",id);
    return i;
  }

  /**
   * 批量导入es
   */
  @Override
  public void exportEs() {
    try {
      // 获取 Flink 的流处理执行环境
      // StreamExecutionEnvironment 是 Flink 流处理程序的上下文环境，用于定义和执行数据流任务
      StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

      BatchESSink batchESSink = new BatchESSink();
      MySQLSource mySQLSource = new MySQLSource();
      // 设置并行度为 1
      // 并行度决定了 Flink 作业中每个算子的并发执行数量，这里设置为 1 表示单线程执行
      env.setParallelism(1);

      // 添加数据源（Source）
      // 使用 MySQLSource 作为数据源，从 MySQL 数据库中读取 YaHighInterview 数据流
      DataStreamSource<YaHighInterview> source = env.addSource(mySQLSource);

      // 添加数据输出（Sink）
      // 将从 MySQL 读取的数据流通过 BatchESSink 输出到 Elasticsearch 中
      source.addSink(batchESSink);

      // 执行 Flink 作业
      // 调用 execute 方法启动 Flink 流处理作业，作业名称为 "MySQL to ES Export with Flink"
      env.execute("MySQL to ES Export with Flink");
    } catch (Exception e) {
      // 捕获异常并抛出运行时异常
      throw new ServiceException("全量导出到es异常");
    }
  }

  @Override
  public int queryExportEsProgress() {
    return MySQLSource.getPercentage();
  }

  /**
   * 查询课程设计关联的高频面试题
   *
   * @param courseDesignId 课程设计id
   * @param courseContent 课程内容
   * @return 高频面试题
   */
  @Override
  public Map<String, Object> queryRelatedInterview(Long courseDesignId, String courseContent) {
    Map<String, Object> result = new HashMap<>();

    // 1. 提取关键词（简单的关键词提取，实际项目中可以使用NLP技术）
    List<String> keywords = extractKeywords(courseContent);

    // 2. 查询高相关度面试题
    List<YaHighInterview> highInterviewList = queryHighRelevanceInterviews(courseContent, keywords);

    // 3. 查询已关联的面试题ID
    List<Long> relatedInterviewIdList = new ArrayList<>();
    if (courseDesignId != null) {
      relatedInterviewIdList = yaCourseDesignInterviewMapper.selectInterviewIdsByCdId(courseDesignId);
    }

    result.put("highInterviewList", highInterviewList);
    result.put("keywords", keywords);
    result.put("relatedInterviewIdList", relatedInterviewIdList);

    return result;
  }

  /**
   * 根据关键词查询课程设计关联的高频面试题
   *
   * @param keywords 关键词
   * @return 高频面试题
   */
  @Override
  public List<YaHighInterview> queryRelatedInterviewByKeywords(List<String> keywords) {
    if (keywords == null || keywords.isEmpty()) {
      return new ArrayList<>();
    }

    List<YaHighInterview> result = new ArrayList<>();

    // 根据每个关键词查询面试题
    for (String keyword : keywords) {
      YaHighInterview query = new YaHighInterview();
      query.setQuestion(keyword);
      List<YaHighInterview> list = selectYaHighInterviewList(query);
      result.addAll(list);
    }

    // 去重
    return result.stream().distinct().collect(Collectors.toList());
  }

  /**
   * 保存课程设计关联的高频面试题
   *
   * @param courseDesignId 课程设计id和面试题id
   * @return 高频面试题
   */
  @Override
  public void saveRelatedInterview(Long courseDesignId, List<Long> interviewIds) {
    // 1. 删除原有的关联关系
    yaCourseDesignInterviewMapper.deleteByCdId(courseDesignId);

    // 2. 插入新的关联关系
    if (interviewIds != null && !interviewIds.isEmpty()) {
      List<YaCourseDesignInterview> relList = new ArrayList<>();
      for (Long interviewId : interviewIds) {
        YaCourseDesignInterview rel = new YaCourseDesignInterview();
        rel.setCdId(courseDesignId);
        rel.setInterviewId(interviewId);
        relList.add(rel);
      }
      yaCourseDesignInterviewMapper.insertBatch(relList);
    }
  }

  /**
   * 提取关键词
   *
   * @param content 内容
   * @return 关键词列表
   */
  private List<String> extractKeywords(String content) {
    List<String> keywords = new ArrayList<>();

    if (content == null || content.trim().isEmpty()) {
      return keywords;
    }

    // 简单的关键词提取方法，实际项目中可以使用更复杂的NLP技术
    // 这里只是示例，提取课程内容中的重要词汇
    String[] words = content.split("[\\s,，;；.。!！?？]+");


    for (String word : words) {
      String cleanWord = word.trim();
      if (cleanWord.length() >= 2 && !keywords.contains(cleanWord)) {
        keywords.add(cleanWord);
      }
    }

    return keywords;
  }
  /**
   * 查询高相关度面试题
   *
   * @param courseContent 课程内容
   * @param keywords      关键词列表
   * @return 高相关度面试题列表
   */
  private List<YaHighInterview> queryHighRelevanceInterviews(String courseContent, List<String> keywords) {
    List<YaHighInterview> result = new ArrayList<>();

    // 根据关键词查询面试题
    if (keywords != null && !keywords.isEmpty()) {
      for (String keyword : keywords) {
        YaHighInterview query = new YaHighInterview();
        query.setQuestion(keyword);
        List<YaHighInterview> list = selectYaHighInterviewList(query);
        result.addAll(list);
      }
    }

    // 根据课程内容查询面试题
    YaHighInterview query = new YaHighInterview();
    query.setQuestion(courseContent);
    List<YaHighInterview> list = selectYaHighInterviewList(query);
    result.addAll(list);

    // 去重并限制数量
    return result.stream()
            .distinct()
            .limit(50) // 限制返回数量
            .collect(Collectors.toList());
  }
}
